package org.zhq.impl.async;

import org.zhq.core.Frame;
import org.zhq.core.IoArgs;
import org.zhq.core.SendPacket;
import org.zhq.core.ds.BytePriorityNode;
import org.zhq.frames.*;

import java.io.Closeable;
import java.io.IOException;

/**
 * 异步包reader
 *
 * @author zhengquan
 * @date 2019/2/9
 */
public class AsyncPacketReader implements Closeable {

    private final PacketProvider provider;
    private volatile IoArgs args = new IoArgs();
    //Frame队列
    private volatile BytePriorityNode<Frame> node;
    private volatile int nodeSize = 0;
    //1,2,3,4,.....255
    private short lastIdentifier = 0;

    public AsyncPacketReader(PacketProvider provider) {
        this.provider = provider;
    }


    boolean requestSendHeartbeatFrame() {
        synchronized (this) {
            for (BytePriorityNode<Frame> x = node; x != null; x = x.next) {
                Frame frame = x.item;
                if (frame.getBodyType() == Frame.TYPE_COMMAND_HEARTBEAT) {
                    return false;
                }
            }
            appendNewFrame(new HeartbeatSendFrame());
            return true;
        }
    }

    /**
     * 请求队列中拿一份Packet进行发送
     * 如果当前reader中有可以用于网络发送的数据,则返回true
     *
     * @return
     */
    boolean requestTakePacket() {
        //还有帧没消费完
        synchronized (this) {
            if (nodeSize >= 1) {
                return true;
            }
        }
        SendPacket packet = provider.takePacket();
        if (packet != null) {
            short identifier = generateIdentifier();
            SendHeaderFrame frame = new SendHeaderFrame(identifier, packet);
            appendNewFrame(frame);
        }
        //可能没有需要发送的包了,连头帧都没有
        synchronized (this) {
            return nodeSize != 0;
        }
    }

    /**
     * 填充数据到IoArgs中
     * 如果当前有可用于发送的帧,则填充数据并返回,如果填充失败可返回null
     *
     * @return
     */
    IoArgs fillData() {
        Frame curFrame = getCurrentFrame();
        if (curFrame == null) {
            return null;
        }
        try {
            if (curFrame.handle(args)) {
                //消费完本帧
                //尝试基于本帧构建后续帧
                Frame nextFrame = curFrame.nextFrame();
                if (nextFrame != null) {
                    appendNewFrame(nextFrame);
                } else if (curFrame instanceof SendEntityFrame) {
                    //末尾实体帧
                    //通知完成
                    provider.completedPacket(((SendEntityFrame) curFrame).getPacket(), true);
                }
                //从链头弹出
                popCurrentFrame();
            }
            return args;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 取消Packet对应的帧发送,如果当前Packet已发送部分数据(就算只是头数据)
     * 也应该在当前帧队列中发送一份取消发送的标志
     *
     * @param packet
     */
    synchronized void cancel(SendPacket packet) {
        if (nodeSize == 0) {
            return;
        }
        for (BytePriorityNode<Frame> x = node, before = null; x != null; before = x, x = x.next) {
            Frame frame = x.item;
            if (frame instanceof AbsSendPacketFrame) {
                AbsSendPacketFrame packetFrame = (AbsSendPacketFrame) frame;
                if (packetFrame.getPacket() == packet) {
                    boolean removable = packetFrame.abort();
                    if (removable) {
                        removeFrame(x, before);
                        if (packetFrame instanceof SendHeaderFrame) {
                            //头帧,并且未被发送任何数据,直接取消后不需要添加取消发送帧
                            break;
                        }
                    }

                    //添加终止帧,通知接收方
                    CancelSendFrame cancelSendFrame = new CancelSendFrame(packetFrame.getBodyIdentifier());
                    appendNewFrame(cancelSendFrame);
                    //意外终止,返回失败
                    provider.completedPacket(packet, false);
                    break;
                }
            }
        }
    }


    private short generateIdentifier() {
        short identifier = ++lastIdentifier;
        if (identifier == 255) {
            lastIdentifier = 0;
        }
        return identifier;
    }

    private synchronized void appendNewFrame(Frame frame) {
        BytePriorityNode<Frame> newFrame = new BytePriorityNode<>(frame);
        if (node != null) {
            node.appendWithPriority(newFrame);
        } else {
            node = newFrame;
        }
        nodeSize++;
    }

    private synchronized Frame getCurrentFrame() {
        if (node == null) {
            return null;
        }
        return node.item;
    }

    private synchronized void popCurrentFrame() {
        node = node.next;
        nodeSize--;
        if (node == null) {
            requestTakePacket();
        }
    }

    private synchronized void removeFrame(BytePriorityNode<Frame> removeNode, BytePriorityNode<Frame> before) {

        if (before == null) {
            node = removeNode.next;
        } else {
            before.next = removeNode.next;
        }
        nodeSize--;
        if (node == null) {
            requestTakePacket();
        }
    }

    @Override
    public synchronized void close() {
        while (node != null) {
            Frame frame = node.item;
            if (frame instanceof AbsSendPacketFrame) {
                SendPacket packet = ((AbsSendPacketFrame) frame).getPacket();
                provider.completedPacket(packet, false);
            }
            node = node.next;
        }
        nodeSize = 0;
        node = null;
    }

    /**
     * 包信息的提供者
     */
    interface PacketProvider {
        /**
         * 获取一个发送包信息
         *
         * @return
         */
        SendPacket takePacket();

        /**
         * 完成包信息发送的回调
         *
         * @param packet
         * @param succeed
         */
        void completedPacket(SendPacket packet, boolean succeed);
    }

}
